#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <errno.h>
#include <string.h>
#include <netdb.h>

#include "iotaimp.h"
#include "zlog.h"
#include "iniparser.h"

// external
PFN_EXTERNAL_CALLBACK_HANDLER g_external_interface1 = NULL;
extern ob_context g_mqtt_ob;
int g_mqtt_issucc = -1;
int g_LocalMqttConnected = -1; // 本地MQTT服务是否连接成功
int g_mqtt_conn = 0;
int sendMessage_errortime = 1;
extern char g_ServerIp[64];
extern char g_ServerPort[64];
int smc_tls = -1;
// callback
PFN_CALLBACK_HANDLER_SAG g_pfnCbSag = NULL;
PFN_LOG_CALLBACK_HANDLER g_pfnLogCb = NULL;
PFN_CALLBACK_HANDLER g_pfnCbConnectSuccess = NULL;
PFN_CALLBACK_HANDLER g_pfnCbConnectFailure = NULL;
PFN_CALLBACK_HANDLER g_pfnCbDisconnectSuccess = NULL;
PFN_CALLBACK_HANDLER g_pfnCbDisconnectFailure = NULL;
PFN_CALLBACK_HANDLER g_pfnCbConnectionLost = NULL;
PFN_CALLBACK_HANDLER g_pfnCbPublishSuccess = NULL;
PFN_CALLBACK_HANDLER g_pfnCbPublishFailure = NULL;
PFN_CALLBACK_HANDLER g_pfnCbSubscribeSuccess = NULL;
PFN_CALLBACK_HANDLER g_pfnCbSubscribeFailure = NULL;
PFN_CALLBACK_HANDLER g_pfnCbCommandArrived = NULL;
PFN_CALLBACK_HANDLER g_pfnCbDeviceAdditionResult = NULL;
PFN_CALLBACK_HANDLER g_pfnCbDeviceDeletionResult = NULL;
PFN_CALLBACK_HANDLER g_pfnCbDeviceUpdateResult = NULL;
PFN_CALLBACK_HANDLER g_pfnCbDeviceQueryResult = NULL;

PFN_CALLBACK_HANDLER_TOPIC g_pfnCallbackTopicHandler = NULL;

char g_pcDeviceId[IOTA_BUF_SIZE_64] = {0};
char g_pcPassword[IOTA_BUF_SIZE_64] = {0};
char g_pcUserName[IOTA_BUF_SIZE_64] = {0};
char g_pcClientId[IOTA_BUF_SIZE_64] = {0};
char g_pcServerAddr[IOTA_BUF_SIZE_64] = {0};
char g_pcCertPath[IOTA_BUF_SIZE_512] = {0};

HW_UINT g_uiPort = 0;
HW_UINT g_uiUrlPrefix = 0;
HW_UINT g_uiAuthMode = 0;
HW_UINT g_uiLogLocalNumber = 0;
HW_UINT g_uiLogLevel = 0;
HW_UINT g_uiKeepaliveTime = 0;
HW_UINT g_uiConnectTimeout = 0;
HW_UINT g_uiRetryInterval = 0;
HW_UINT g_uiQos = 0;
HW_UINT g_uiSSL = 0;
char g_pcTopicTopoAdd[IOTA_BUF_SIZE_128] = {0};
char g_pcTopicTopoAddResponse[IOTA_BUF_SIZE_128] = {0};
char g_pcTopicTopoUpdate[IOTA_BUF_SIZE_128] = {0};
char g_pcTopicTopoUpdateResponse[IOTA_BUF_SIZE_128] = {0};
char g_pcTopicTopoDelete[IOTA_BUF_SIZE_128] = {0};
char g_pcTopicTopoQuery[IOTA_BUF_SIZE_128] = {0};
char g_pcTopicTopoQueryResponse[IOTA_BUF_SIZE_128] = {0};
char g_pcTopicCommand[IOTA_BUF_SIZE_128] = {0};
char g_pcTopicCommandResponse[IOTA_BUF_SIZE_128] = {0};
char g_pcTopicDatas[IOTA_BUF_SIZE_128] = {0};

// mqtt
pthread_mutex_t g_clientMutex;
MQTTAsync g_client = NULL;
MQTTAsync g_LocalMqttClient = NULL;
char *g_mqtt_module[] = {"device", "network", "virtualization", "app", "security", "system", "diagnostics", "trace", "ota", "dictionary", "third","plc"};
char *g_mqtt_operation[] = {"set", "get", "command"};
char g_topic_arr[40][128];
int g_topic_num = 0;
extern int PRIORITY_CONF;

char g_pcWorkPath[IOTA_BUF_SIZE_512] = {0};
char g_pcLogPath[IOTA_BUF_SIZE_512] = {0};

zlog_category_t *BD_zlog = NULL;
zlog_category_t *PLCC_zlog = NULL;

static void subscribe(void);

// static fun_recv pFunMB = NULL;
// static fun_recv pFunLSM = NULL;

int IOTA_sendMqtt(char *pcTopicName, char *buf, int size)
{
	return publishMessage(pcTopicName, buf);
}

int IOTA_sendToLocalMqtt(char *pcTopicName, char *buf)
{
	return publishLocalMessage(pcTopicName, buf);
}

static int SetMqttOnline(int iStatus)
{
	// 先加载配置文件复本
	char sPath[255] = "";
	snprintf(sPath, sizeof(sPath), "%s.tmp", EagleEyeSMG_INI_DIR);
	// dzlog_info("path:%s",sPath);
	dictionary *pRootInit = iniparser_load(sPath);
	if (pRootInit)
	{
		// 获取MQTT连接在线状态值,如果相同则不需要重新设置，不同则继续往下执行
		int iRet = iniparser_getint(pRootInit, "version:online", 0);
		if (iRet == iStatus)
		{
			iniparser_freedict(pRootInit);
			dzlog_info("current online status:%d", iStatus);
			return 0;
		}
	}
	else
	{
		dzlog_warn("EagleEyeSMG_INI_DIR tmp file don't exist!");

		// 创建一个配置文件复本
		FILE *pFile = fopen(sPath, "wt+");
		if (pFile == NULL)
		{
			dzlog_error("create EagleEyeSMG_INI_DIR tmp file failed!\n");
			return -1;
		}

		char sBuf[32] = "[version]\nonline = 0\n";
		fwrite(sBuf, 1, strlen(sBuf), pFile);
		fclose(pFile);
		dzlog_info("create EagleEyeSMG_INI_DIR tmp file success\n");

		// 再次加载配置文件
		pRootInit = iniparser_load(sPath);
	}

	if (pRootInit)
	{
		// 设置MQTT连接在线状态值
		char buf[4] = {0};
		sprintf(buf, "%d", iStatus);
		int ret = iniparser_set(pRootInit, "version:online", buf);
		if (ret == -1)
		{
			dzlog_error("set online to ini file failed!");
			iniparser_freedict(pRootInit);
			return -1;
		}

		// 打开配置文件复本
		FILE *pFile = fopen(sPath, "wt+");

		// 将内容写入文件
		iniparser_dump_ini(pRootInit, pFile);
		iniparser_freedict(pRootInit);
		fclose(pFile);
		dzlog_info("SetMqttOnline status:%d", iStatus);
		return 0;
	}
	else
	{
		dzlog_error("load EagleEyeSMG_INI_DIR tmp file failed!\n");
		return -1;
	}
}

static void writeLog(int level, char *format, ...)
{
	// if (g_pfnLogCb)
	//{
	//	va_list vl;
	//	va_start(vl, format);
	//	g_pfnLogCb(level, format, vl);
	// }
}

static void delivered(void *context, MQTTAsync_token dt)
{
	dzlog_debug("Message with token value %d delivery confirmed\n", dt);
}
void IOTA_setCallbackSag(PFN_CALLBACK_HANDLER_SAG pFun)
{
	if (pFun)
		g_pfnCbSag = pFun;
}

static int msgarrvd(void *context, char *topicName, int topicLen, MQTTAsync_message *message)
{
	int iRet = -1;
	//((char*)message->payload)[message->payloadlen] = '\0';

	// HW_BOOL bRet = 0;
	// HW_UINT uiMid = 0;
	// int iStatusCode = 0;
	// char pcStatusDecs[IOTA_BUF_SIZE_64] = { 0 };
	if (g_pfnCbSag)
	{
		g_pfnCbSag(topicName, (char *)message->payload, message->payloadlen);
	}

	MQTTAsync_freeMessage(&message);
	MQTTAsync_free(topicName);

	return iRet;
}

static int LocalMqttMsgCB(void *context, char *topicName, int topicLen, MQTTAsync_message *message)
{
	int iRet = -1;
	// if (g_pfnCbSag)
	// {
	// 	g_pfnCbSag(topicName, (char *)message->payload, message->payloadlen);
	// }

	MQTTAsync_freeMessage(&message);
	MQTTAsync_free(topicName);
	return iRet;
}

/**
 * 将错误代码转换为可读字符串
 */
const char* connectivity_error_str(int error) {
	switch (error) {
	case 0:  return "连接成功";
	case -1: return "参数错误";
	case -2: return "创建socket失败";
	case -3: return "设置非阻塞失败";
	case -4: return "连接失败";
	case -5: return "连接超时";
	case -6: return "系统错误";
	default: return "未知错误";
	}
}

static void* self_thread(void* param)
{
	int result = check_connectivity(g_ServerIp, atoi(g_ServerPort) , 5);
	char res[255] = { 0 };
	if (result == 0) {
		sprintf(res, "%s", "网络检测: 服务器端口可达，可能是MQTT服务问题");
		dzlog_error(res);
	}
	else {
		sprintf(res, "网络检测: 连接失败(%d: %s)", result, connectivity_error_str(result));
		dzlog_error(res);

		//查询设备当前的防火墙与路由表

		//进行telnet测试

	}



	return;
}

static void connlost(void *context, char *cause)
{
	g_mqtt_conn = 0;
	g_mqtt_issucc = 0;
	dzlog_error("Connection lost! cause: %s\n", cause);
	printf("Connection lost! cause: %s\n", cause);

	if (g_pfnCbConnectionLost)
	{
		g_pfnCbConnectionLost(NULL, 0, 0, cause);
	}
	
	//起线程检测网络情况是否正常，并记录日志作为证据
	pthread_t tid_checking;
	int ret = pthread_create(&tid_checking, NULL, self_thread, NULL);
	if (ret != 0)
	{
		dzlog_error("pthread_create self_thread error\n");
		// return;
	}
	setstatus(0);

}

static void LocalConnLost(void *context, char *cause)
{
	g_LocalMqttConnected = 0;
	printf("LocalConnLost lost! cause: %s\n", cause);
	dzlog_error("LocalConnLost lost! cause: %s", cause);
	// if (g_pfnCbConnectionLost)
	// {
	// 	g_pfnCbConnectionLost(NULL, 0, 0, cause);
	// }
}

static void onConnectSuccess(void *context, MQTTAsync_successData *response)
{
	printf("onConnectSuccess! token: %d\n", response->token);
	dzlog_info("onConnectSuccess! token: %d\n", response->token);
	//if (g_mqtt_conn == 0)
	//{
	//	g_mqtt_conn = 1;
	//	if (g_pfnCbConnectSuccess)
	//	{
	//		g_pfnCbConnectSuccess(NULL, 0, 0, "");
	//	}
	//	subscribe();
	//}
	//setstatus(1);
}

static void onLocalConnectSuccess(void *context, MQTTAsync_successData *response)
{
	// printf("onLocalConnectSuccess: token: %d\n", response->token);
	dzlog_info("onLocalConnectSuccess: token: %d", response->token);
}

static void onConnectFailure(void *context, MQTTAsync_failureData *response)
{
	g_mqtt_conn = 0;
	setstatus(0);
	dzlog_error("onConnectFailure! token: %d, code: %d, message = %s\n", response->token, response->code, response->message);
	g_mqtt_issucc = -2;
	if (response->code == 5 || response->code == 4)
	{
		printf("onConnectFailure! mqtt Connection ip[%s] Password error\n", g_mqtt_ob.clientInfo.server_addr);
		dzlog_error("onConnectFailure! mqtt Connection ip[%s] Password error", g_mqtt_ob.clientInfo.server_addr);
		g_mqtt_issucc = -5;
	}

	//    ʧ ܻص
	if (g_pfnCbConnectFailure)
	{
		g_pfnCbConnectFailure(NULL, 0, response->code, (char *)response->message);
	}

}

static void onLocalConnectFailure(void *context, MQTTAsync_failureData *response)
{
	printf("onLocalConnectFailure: token: %d, code: %d, message = %s\n", response->token, response->code, response->message);
	dzlog_error("onLocalConnectFailure: token: %d, code: %d, message = %s", response->token, response->code, response->message);
	g_LocalMqttConnected = -2;
	if (response->code == 5 || response->code == 4)
	{
		dzlog_error("onLocalConnectFailure: mqtt Connection local Password error");
		g_LocalMqttConnected = -5;
	}
}

static void onSubscribeSuccess(void *context, MQTTAsync_successData *response)
{
	// dzlog_info("onSubscribeSuccess! token: %d\n", response->token);

	if (g_pfnCbSubscribeSuccess)
	{
		g_pfnCbSubscribeSuccess(NULL, 0, 0, "");
	}
}

static void onLocalSubscribeSuccess(void *context, MQTTAsync_successData *response)
{
	// printf("onLocalSubscribeSuccess(token: %d)\n", response->token);
	// dzlog_info("onLocalSubscribeSuccess(token: %d)\n", response->token);
}

static void onSubscribeFailure(void *context, MQTTAsync_failureData *response)
{
	printf("onSubscribeFailure! token: %d, code: %d, message = %s\n", response->token, response->code, response->message);
	dzlog_error("onSubscribeFailure! token: %d, code: %d, message = %s\n", response->token, response->code, response->message);
	if (g_pfnCbSubscribeFailure)
	{
		g_pfnCbSubscribeFailure(NULL, 0, response->code, (char *)response->message);
	}
}

static void onLocalSubscribeFailure(void *context, MQTTAsync_failureData *response)
{
	printf("onLocalSubscribeFailure(token: %d, code: %d, message = %s)\n", response->token, response->code, response->message);
	dzlog_error("onLocalSubscribeFailure(token: %d, code: %d, message = %s)", response->token, response->code, response->message);
}

static void onDisconnectSuccess(void *context, MQTTAsync_successData *response)
{
	dzlog_info("onDisconnectSuccess! token: %d\n", response->token);

	if (g_pfnCbDisconnectSuccess)
	{
		g_pfnCbDisconnectSuccess(NULL, 0, 0, "");
	}
}

static void onDisconnectFailure(void *context, MQTTAsync_failureData *response)
{
	dzlog_error("onDisconnectFailure! token: %d, code: %d, message = %s\n", response->token, response->code, response->message);

	if (g_pfnCbDisconnectFailure)
	{
		g_pfnCbDisconnectFailure(NULL, 0, response->code, (char *)response->message);
	}
}

static void onSendSuccess(void *context, MQTTAsync_successData *response)
{
	if (g_pfnCbPublishSuccess)
	{
		g_pfnCbPublishSuccess(NULL, 0, 0, "");
	}
}

static void onSendLocalSuccess(void *context, MQTTAsync_successData *response)
{
}

static void onSendFailure(void *context, MQTTAsync_failureData *response)
{
	dzlog_error("onSendFailure! token: %d, code: %d", response->token, response->code);

	if (g_pfnCbPublishFailure)
	{
		if (response->message)
		{
			g_pfnCbPublishFailure(NULL, 0, response->code, (char*)response->message);
		}
		else
		{
			g_pfnCbPublishFailure(NULL, 0, response->code, NULL);
		}
	}
}

static void onSendLocalFailure(void *context, MQTTAsync_failureData *response)
{
	printf("onSendLocalFailure! token: %d, code: %d, message = %s\n", response->token, response->code, response->message);
	dzlog_error("onSendLocalFailure! token: %d, code: %d, message = %s", response->token, response->code, response->message);
}

static void onConnectCallCBack(void *context, char *cause)
{
	dzlog_info("Successful onConnectCallCBack cause[%s]", cause);

	if (g_mqtt_conn == 0)
	{
		g_mqtt_conn = 1;
		if (g_pfnCbConnectSuccess)
		{
			g_pfnCbConnectSuccess(NULL, 0, 0, "");
		}
		subscribe();
	}
	if (g_mqtt_conn == 1)
	{
		setstatus(1);
		g_mqtt_issucc = 1;
		if (g_mqtt_ob.clientInfo.SSL == 2) // 如果启用国密，无论证书是否加密，使用完后都得删除该临时目录下的证书文件
		{
			FILE* fp = NULL;
			fp = popen("rm -rf /userdata/security/.certificate/*", "r");
			pclose(fp);
		}

	}
}

static void onLocalConnectCB(void *context, char *cause)
{
	printf("onLocalConnectCB success\n");
	dzlog_info("onLocalConnectCB success.");
	g_LocalMqttConnected = 1;

	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	opts.onSuccess = onLocalSubscribeSuccess;
	opts.onFailure = onLocalSubscribeFailure;
	opts.context = NULL;

	int iRet = MQTTAsync_subscribe(g_LocalMqttClient, "PLCService/notify/spont/*/accessNetDevAlaram", g_uiQos, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		dzlog_error("Failed to subscribe alarm(error: %d)!", iRet);
	}
	else
		dzlog_info("subscribe local alarm success");

	iRet = MQTTAsync_subscribe(g_LocalMqttClient, "/ext.syy.phSmc/ext.syy.mccb/S-smclink/F-ReqRandom", g_uiQos, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		dzlog_error("Failed to subscribe F-ReqRandom(error: %d)!", iRet);
	}
	else
		dzlog_info("subscribe F-ReqRandom success");

	iRet = MQTTAsync_subscribe(g_LocalMqttClient, "/ext.syy.phSmc/SyyMccb/S-smclink/F-ReqRandom", g_uiQos, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		dzlog_error("Failed to subscribe F-ReqRandom(error: %d)!", iRet);
	}
	else
		dzlog_info("subscribe F-ReqRandom success");

	iRet = MQTTAsync_subscribe(g_LocalMqttClient, "/ext.syy.phSmc/ext.syy.mccb/S-smclink/F-VerifySign", g_uiQos, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		dzlog_error("Failed to subscribe F-VerifySign(error: %d)!", iRet);
	}
	else
		dzlog_info("subscribe F-VerifySign success");

	iRet = MQTTAsync_subscribe(g_LocalMqttClient, "/ext.syy.phSmc/SyyMccb/S-smclink/F-VerifySign", g_uiQos, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		dzlog_error("Failed to subscribe F-VerifySign(error: %d)!", iRet);
	}
	else
		dzlog_info("subscribe F-VerifySign success");

	iRet = MQTTAsync_subscribe(g_LocalMqttClient, "/ext.syy.phSmc/ext.syy.mccb/S-smclink/F-ReportDeviceInfo", g_uiQos, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		dzlog_error("Failed to subscribe F-ReportDeviceInfo(error: %d)!", iRet);
	}
	else
		dzlog_info("subscribe F-ReportDeviceInfo success");

	iRet = MQTTAsync_subscribe(g_LocalMqttClient, "/ext.syy.phSmc/SyyMccb/S-smclink/F-ReportDeviceInfo", g_uiQos, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		dzlog_error("Failed to subscribe F-ReportDeviceInfo(error: %d)!", iRet);
	}
	else
		dzlog_info("subscribe F-ReportDeviceInfo success");

	iRet = MQTTAsync_subscribe(g_LocalMqttClient, "/sys.brd/ext.syy.mccb/S-dataservice/F-UpdateDevStatus", g_uiQos, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		dzlog_error("Failed to subscribe F-UpdateDevStatus(error: %d)!", iRet);
	}
	else
		dzlog_info("subscribe F-UpdateDevStatus success");

	iRet = MQTTAsync_subscribe(g_LocalMqttClient, "SyyMccb/notify/spont/*/DC_MCCB/#", g_uiQos, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		dzlog_error("Failed to subscribe F-UpdateDevStatus(error: %d)!", iRet);
	}
	else
		dzlog_info("subscribe F-UpdateDevStatus success");

	iRet = MQTTAsync_subscribe(g_LocalMqttClient, "/LocalMqtt/BianDian/V1/Cloud_BdPortal_Res/+", g_uiQos, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		dzlog_error("Failed to subscribe Cloud_BdPortal_Res(error: %d)!", iRet);
	}
	else
	{
		dzlog_info("subscribe Cloud_BdPortal_Res success");
		// 针对变电的日志
		BD_zlog = zlog_get_category("GetBD");
		zlog_debug(BD_zlog, "Local MQTT start");
	}


	//TODO-YZ:PLCC 注册
	iRet = MQTTAsync_subscribe(g_LocalMqttClient, "/ext.syy.phSmc/ext.syy.plcc/S-smclink/F-PlccEvent", g_uiQos, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		dzlog_error("Failed to subscribe F-PlccEvent(error: %d)!", iRet);
	}
	else
	{
		dzlog_info("subscribe F-PlccEvent success");
		// 针对变电的日志
		PLCC_zlog = zlog_get_category("GetPLCC");
		zlog_debug(PLCC_zlog, "Local MQTT-PLCC start");
	}
	
	//跟采集调度间的主题
	iRet = MQTTAsync_subscribe(g_LocalMqttClient, "/ext.syy.phSmc/svc.tasksched/S-meter/F-ProxyFrame", g_uiQos, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		dzlog_error("Failed to subscribe F-ProxyFrame(error: %d)!", iRet);
	}
	else
		dzlog_info("subscribe F-ProxyFrame success");

	iRet = MQTTAsync_subscribe(g_LocalMqttClient, "/svc.brd/svc.tasksched/S-meter/F-ReportEvent", g_uiQos, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		dzlog_error("Failed to subscribe svc.tasksched/S-meter/F-ReportEvent(error: %d)!", iRet);
	}
	else
		dzlog_info("subscribe svc.tasksched/S-meter/F-ReportEvent success");
}

int publishMessage(char *pcTopicName, char *pcMessage)
{
	if (g_mqtt_issucc != 1)
	{
		printf("Failed to publish message,MQTTAsync_isDisConnected, return code %d\n", -3);
		dzlog_info("Failed to publish message,MQTTAsync_isDisConnected, return code %d\n", -3);
		sendMessage_errortime++;
		if (0 == sendMessage_errortime % 60)
		{
			int result = check_connectivity(g_ServerIp, atoi(g_ServerPort), 5);
			char res[255] = { 0 };
			if (result == 0) {
				sprintf(res, "%s", "Network detection:[%s:%s] is reachable, possibly an MQTT service issue", g_ServerIp, g_ServerPort);
				dzlog_error(res);
				dzlog_error("restart EagleEyeSMG");
				exit(0);
			}
			else {
				sprintf(res, "Network detection:[%s:%s] Connection failed(%d: %s)", g_ServerIp, g_ServerPort, result, connectivity_error_str(result));
				dzlog_error(res);
			}
			dzlog_error("publishMessage g_mqtt_issucc sendMessage_errortime more than 60 times.");
			usleep(200 * 1000);
		}
		//if (sendMessage_errortime > 250)
		//{
		//	// 程序自杀，
		//	dzlog_error("publishMessage g_mqtt_issucc sendMessage_errortime more than 250 times. restart EagleEyeSMG");
		//	exit(0);
		//}
		return IOTA_FAILURE;
	}

	int iRet = IOTA_SUCCESS;
	MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
	MQTTAsync_token token;

	pubmsg.payload = (void *)pcMessage;
	pubmsg.payloadlen = strlen(pcMessage);
	pubmsg.qos = g_uiQos;
	pubmsg.retained = 0;

	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	if(g_mqtt_ob.clientInfo.scram == 1)
	{
		opts.onSuccess5 = onSendSuccess;
		opts.onFailure5 = onSendFailure;
	}
	else
	{
		opts.onSuccess = onSendSuccess;
		opts.onFailure = onSendFailure;
	}
	opts.context = NULL;

	pthread_mutex_lock(&g_clientMutex);
	iRet = MQTTAsync_sendMessage(g_client, pcTopicName, &pubmsg, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		dzlog_error("Failed to publish message,topic: %s, return code %d", pcTopicName, iRet);
		if (iRet == MQTTASYNC_PERSISTENCE_ERROR)
		{
			dzlog_info("redo to publish message,topic: %s", pcTopicName);
			sleep(1);
			iRet = MQTTAsync_sendMessage(g_client, pcTopicName, &pubmsg, &opts);
		}
		if (iRet != MQTTASYNC_SUCCESS)
		{
			iRet = IOTA_FAILURE;
			sendMessage_errortime++;
			if (0 == sendMessage_errortime % 60)
			{
				int result = check_connectivity(g_ServerIp, atoi(g_ServerPort), 5);
				char res[255] = { 0 };
				if (result == 0) {
					sprintf(res, "%s", "Network detection:[%s:%s] is reachable, possibly an MQTT service issue", g_ServerIp, g_ServerPort);
					dzlog_error(res);
					exit(0);
				}
				else {
					sprintf(res, "Network detection:[%s:%s] Connection failed(%d: %s)", g_ServerIp, g_ServerPort, result, connectivity_error_str(result));
					dzlog_error(res);
				}
				dzlog_error("Sending messages failed more than 60 times");
				usleep(200 * 1000);
			}
			//if (sendMessage_errortime > 250)
			//{
			//	// 程序自杀，
			//	dzlog_error("Sending messages failed more than 250 times.restart EagleEyeSMG");
			//	exit(0);
			//}
		}
		else
			dzlog_info("redo MQTTAsync_sendMessage succ, topic: %s, msg: %s", pcTopicName, pcMessage);
	}
	else
	{
		// 使用 const 变量定义目标字符串（替代 #define）
		const char *const target = "/v1/devices/test/netTestRequest";
		// 编译时计算目标字符串长度（不包含终止符）
		const size_t target_len = sizeof("/v1/devices/test/netTestRequest") - 1;
		if (strncmp(pcTopicName, target, target_len) == 0 && pcTopicName[target_len] == '\0') // 确保长度相同
		{
			//
		}
		else
		{
			dzlog_info("MQTTAsync_sendMessage, topic: %s, msg: %s", pcTopicName, pcMessage);
		}
		sendMessage_errortime = 1;
	}
	pthread_mutex_unlock(&g_clientMutex);
	return iRet;
}

int publishLocalMessage(char *pcTopicName, char *pcMessage)
{
	if (g_LocalMqttConnected != 1)
	{
		printf("Failed to publish message to local mqtt:%d\n", g_LocalMqttConnected);
		dzlog_info("Failed to publish message to local mqtt:%d\n", g_LocalMqttConnected);
		return IOTA_FAILURE;
	}

	int iRet = IOTA_SUCCESS;
	MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
	MQTTAsync_token token;

	pubmsg.payload = (void *)pcMessage;
	pubmsg.payloadlen = strlen(pcMessage);
	pubmsg.qos = g_uiQos;
	pubmsg.retained = 0;

	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	opts.onSuccess = onSendLocalSuccess;
	opts.onFailure = onSendLocalFailure;
	opts.context = NULL;
	iRet = MQTTAsync_sendMessage(g_LocalMqttClient, pcTopicName, &pubmsg, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		dzlog_error("Failed to publish message to local,topic: %s, return code %d", pcTopicName, iRet);
		iRet = IOTA_FAILURE;
	}
	else
	{
		dzlog_debug("publish message to local success, topic: %s, msg: %s", pcTopicName, pcMessage);
	}
	return iRet;
}

void set_topic_str()
{
	int moduleNum = 0;
	int operNum = 0;
	int topicNum = 0;

	for (moduleNum = 0; moduleNum < (sizeof(g_mqtt_module) / sizeof(g_mqtt_module[0])); moduleNum++)
	{
		for (operNum = 0; operNum < (sizeof(g_mqtt_operation) / sizeof(g_mqtt_operation[0])); operNum++)
		{
			// dzlog_info("module[%s],oper[%s]\n",g_mqtt_module[moduleNum],g_mqtt_operation[operNum]);
			sprintf(g_topic_arr[topicNum], "%s%s/%s/%s", IOTA_TOPIC_PREFIX, g_pcDeviceId, g_mqtt_module[moduleNum], g_mqtt_operation[operNum]);
			// dzlog_info("topic_str topicNum[%d][%s]\n", topicNum, g_topic_arr[topicNum]);
			topicNum++;
			g_topic_num = g_topic_num + 1;
		}
	}
	printf("g_topic_num[%d]\n", g_topic_num);
	sprintf(g_topic_arr[topicNum], "%s%s/%s/%s", IOTA_TOPIC_PREFIX, g_pcDeviceId, "securityZn", "set");

	g_topic_num++;
}

static void subscribe(void)
{
	int iRet;
	int num_t = 0;
	char netTestTopic[128] = {0};
	sprintf(netTestTopic, "%s%s%s", IOTA_TOPIC_PREFIX, g_pcDeviceId, "/test/netTestResponse");
	dzlog_info("netTest subscribe, topic: %s", netTestTopic);

	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	if (g_mqtt_ob.clientInfo.scram == 1)
	{
		opts.onSuccess5 = onSubscribeSuccess;
		opts.onFailure5 = onSubscribeFailure;
	}
	else
	{
		opts.onSuccess = onSubscribeSuccess;
		opts.onFailure = onSubscribeFailure;
	}
	opts.context = NULL;

	iRet = IOTA_SUCCESS;
	{
		// demo

		iRet = MQTTAsync_subscribe(g_client, netTestTopic, g_uiQos, &opts);
		if (iRet != MQTTASYNC_SUCCESS)
		{
			dzlog_error("Failed to subscribe, return code %d", iRet);
		}
		else
		{
			num_t++;
			dzlog_info("MQTTAsync_subscribe, topic: %s", netTestTopic);
		}
	}

	char BD_Topic[128] = {0};
	sprintf(BD_Topic, "%s%s/Cloud_BdPortal/+", IOTA_TOPIC_PREFIX, g_pcDeviceId);
	iRet = MQTTAsync_subscribe(g_client, BD_Topic, g_uiQos, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		// dzlog_error( "Failed to subscribe, return code %d\n", iRet);
		dzlog_error("Failed to subscribe, return code %d\n", iRet);
	}
	else
	{
		num_t++;
		// dzlog_error( "MQTTAsync_subscribe, topic: %s\n", arraySubs[i]);
		dzlog_info("MQTTAsync_subscribe, topic: %s\n", BD_Topic);
	}
	//TODO-YZ:PLCC 注册
	char PLCC_Topic[128] = {0};
	sprintf(PLCC_Topic, "%s%s/app/cloudProgram", IOTA_TOPIC_PREFIX, g_pcDeviceId);
	
	iRet = MQTTAsync_subscribe(g_client, PLCC_Topic, g_uiQos, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		// dzlog_error( "Failed to subscribe, return code %d\n", iRet);
		dzlog_error("Failed to subscribe, return code %d", iRet);
	}
	else
	{
		num_t++;
		// dzlog_error( "MQTTAsync_subscribe, topic: %s\n", arraySubs[i]);
		dzlog_info("MQTTAsync_subscribe, topic: %s", PLCC_Topic);
	}
	int i = 0;
	for (; i < g_topic_num; i++)
	{
		iRet = MQTTAsync_subscribe(g_client, g_topic_arr[i], g_uiQos, &opts);
		if (iRet != MQTTASYNC_SUCCESS)
		{
			dzlog_error("Failed to subscribe, return code %d", iRet);
			printf("Failed to subscribe, return code %d", iRet);
		}
		else
		{
			num_t++;
			dzlog_info("MQTTAsync_subscribe, topic: %s", g_topic_arr[i]);
		}
	}
	dzlog_info("-----------------------------------------------subscribe count:%d", num_t);
}

void IOTA_setExternalCallback(int iItem, PFN_CALLBACK_HANDLER pfnCallbackHandler)
{
	switch (iItem)
	{
	case EN_IOTA_CALLBACK_EXTERNAL_INTERFACE1:
		g_external_interface1 = pfnCallbackHandler;
		break;
	}
}

int IOTA_init(char *pcWorkPath, char *pcLogPath)
{
	strncpy(g_pcWorkPath, pcWorkPath, IOTA_BUF_SIZE_512 - 1);
	strncpy(g_pcLogPath, pcLogPath, IOTA_BUF_SIZE_512 - 1);
	return IOTA_SUCCESS;
}

void IOTA_destroy()
{
	if (g_client == NULL)
		return;

	MQTTAsync_destroy(&g_client);
	g_client = NULL;
}

void IOTA_setCallback(int iItem, PFN_CALLBACK_HANDLER pfnCallbackHandler)
{
	switch (iItem)
	{
	case EN_IOTA_CALLBACK_CONNECT_SUCCESS:
		g_pfnCbConnectSuccess = pfnCallbackHandler;
		break;
	case EN_IOTA_CALLBACK_CONNECT_FAILURE:
		g_pfnCbConnectFailure = pfnCallbackHandler;
		break;
	case EN_IOTA_CALLBACK_DISCONNECT_SUCCESS:
		g_pfnCbDisconnectSuccess = pfnCallbackHandler;
		break;
	case EN_IOTA_CALLBACK_DISCONNECT_FAILURE:
		g_pfnCbDisconnectFailure = pfnCallbackHandler;
		break;
	case EN_IOTA_CALLBACK_CONNECTION_LOST:
		g_pfnCbConnectionLost = pfnCallbackHandler;
		break;
	case EN_IOTA_CALLBACK_PUBLISH_SUCCESS:
		g_pfnCbPublishSuccess = pfnCallbackHandler;
		break;
	case EN_IOTA_CALLBACK_PUBLISH_FAILURE:
		g_pfnCbPublishFailure = pfnCallbackHandler;
		break;
	case EN_IOTA_CALLBACK_SUBSCRIBE_SUCCESS:
		g_pfnCbSubscribeSuccess = pfnCallbackHandler;
		break;
	case EN_IOTA_CALLBACK_SUBSCRIBE_FAILURE:
		g_pfnCbSubscribeFailure = pfnCallbackHandler;
		break;
	case EN_IOTA_CALLBACK_COMMAND_ARRIVED:
		g_pfnCbCommandArrived = pfnCallbackHandler;
		break;
	case EN_IOTA_CALLBACK_DEVICE_ADDITION_RESULT:
		g_pfnCbDeviceAdditionResult = pfnCallbackHandler;
		break;
	case EN_IOTA_CALLBACK_DEVICE_DELETION_RESULT:
		g_pfnCbDeviceDeletionResult = pfnCallbackHandler;
		break;
	case EN_IOTA_CALLBACK_DEVICE_UPDATE_RESULT:
		g_pfnCbDeviceUpdateResult = pfnCallbackHandler;
		break;
	case EN_IOTA_CALLBACK_DEVICE_QUERY_RESULT:
		g_pfnCbDeviceQueryResult = pfnCallbackHandler;
		break;
	default:
		break;
	}
}

void IOTA_setCallbackWithTopic(PFN_CALLBACK_HANDLER_TOPIC pfnCallbackTopicHandler)
{
	g_pfnCallbackTopicHandler = pfnCallbackTopicHandler;
}

void IOTA_setPrintLogCallback(PFN_LOG_CALLBACK_HANDLER pfnLogCallbackHandler)
{
	g_pfnLogCb = pfnLogCallbackHandler;
}

int IOTA_configSetStr(int iItem, char *pValue)
{
	int iRet = IOTA_SUCCESS;
	switch (iItem)
	{
	case EN_IOTA_CFG_DEVICEID:
		strncpy(g_pcDeviceId, pValue, IOTA_BUF_SIZE_64 - 1);
		sprintf(g_pcTopicTopoAdd, "%s%s%s", IOTA_TOPIC_PREFIX, g_pcDeviceId, IOTA_TOPIC_POSTFIX_TOPO_ADD);
		sprintf(g_pcTopicTopoAddResponse, "%s%s%s", IOTA_TOPIC_PREFIX, g_pcDeviceId, IOTA_TOPIC_POSTFIX_TOPO_ADDRESPONSE);
		sprintf(g_pcTopicTopoUpdate, "%s%s%s", IOTA_TOPIC_PREFIX, g_pcDeviceId, IOTA_TOPIC_POSTFIX_TOPO_UPDATE);
		sprintf(g_pcTopicTopoUpdateResponse, "%s%s%s", IOTA_TOPIC_PREFIX, g_pcDeviceId, IOTA_TOPIC_POSTFIX_TOPO_UPDATERESPONSE);
		sprintf(g_pcTopicTopoDelete, "%s%s%s", IOTA_TOPIC_PREFIX, g_pcDeviceId, IOTA_TOPIC_POSTFIX_TOPO_DELETE);
		sprintf(g_pcTopicTopoQuery, "%s%s%s", IOTA_TOPIC_PREFIX, g_pcDeviceId, IOTA_TOPIC_POSTFIX_TOPO_QUERY);
		sprintf(g_pcTopicTopoQueryResponse, "%s%s%s", IOTA_TOPIC_PREFIX, g_pcDeviceId, IOTA_TOPIC_POSTFIX_TOPO_QUERYRESPONSE);
		sprintf(g_pcTopicCommand, "%s%s%s", IOTA_TOPIC_PREFIX, g_pcDeviceId, IOTA_TOPIC_POSTFIX_COMMAND);
		sprintf(g_pcTopicCommandResponse, "%s%s%s", IOTA_TOPIC_PREFIX, g_pcDeviceId, IOTA_TOPIC_POSTFIX_COMMANDRESPONSE);
		sprintf(g_pcTopicDatas, "%s%s%s", IOTA_TOPIC_PREFIX, g_pcDeviceId, IOTA_TOPIC_POSTFIX_DATAS);
		set_topic_str();
		break;
	case EN_IOTA_CFG_DEVICESECRET:
		strncpy(g_pcPassword, pValue, IOTA_BUF_SIZE_64 - 1);
		break;
	case EN_IOTA_CFG_USERNAME:
		strncpy(g_pcUserName, pValue, IOTA_BUF_SIZE_64 - 1);
		break;
	case EN_IOTA_CFG_CLIENTID:
		strncpy(g_pcClientId, pValue, IOTA_BUF_SIZE_64 - 1);
		break;
	case EN_IOTA_CFG_MQTT_ADDR:
		strncpy(g_pcServerAddr, pValue, IOTA_BUF_SIZE_64 - 1);
		break;
	default:
		iRet = IOTA_FAILURE;
		break;
	}
	return iRet;
}

int IOTA_configSetUint(int iItem, HW_UINT uiValue)
{
	int iRet = IOTA_SUCCESS;
	switch (iItem)
	{
	case EN_IOTA_CFG_MQTT_PORT:
		g_uiPort = uiValue;
		break;
	case EN_IOTA_CFG_MQTT_URL_PREFIX:
		g_uiUrlPrefix = uiValue;
		break;
	case EN_IOTA_CFG_AUTH_MODE:
		g_uiAuthMode = uiValue;
		break;
	case EN_IOTA_CFG_LOG_LOCAL_NUMBER:
		g_uiLogLocalNumber = uiValue;
		break;
	case EN_IOTA_CFG_LOG_LEVEL:
		g_uiLogLevel = uiValue;
		break;
	case EN_IOTA_CFG_KEEP_ALIVE_TIME:
		g_uiKeepaliveTime = uiValue;
		break;
	case EN_IOTA_CFG_CONNECT_TIMEOUT:
		g_uiConnectTimeout = uiValue;
		break;
	case EN_IOTA_CFG_RETRY_INTERVAL:
		g_uiRetryInterval = uiValue;
		break;
	case EN_IOTA_CFG_QOS:
		g_uiQos = uiValue;
		break;
	case EN_IOTA_CFG_SSL:
		g_uiSSL = uiValue;
		break;
	default:
		iRet = IOTA_FAILURE;
		break;
	}
	return iRet;
}

int mqtt_login(char *ip, int type, char *port, char *user, char *pwd, int SSL)
{
	dzlog_info("mqtt_login start ip[%s],port[%s],SSL[%d]", ip, port, SSL);

	if (SSL == 3)
	{
		g_uiSSL = 2;
		g_mqtt_ob.clientInfo.SSL = 2;
	}
	else
	{
		g_uiSSL = SSL;
		g_mqtt_ob.clientInfo.SSL = SSL;
	}


	int bConnect = -1;
	int iden = 0;
	bool pwdchange = false;
	char pcUrl[IOTA_BUF_SIZE_64] = {0};
	for (iden = 0; iden < 3; iden++)
	{
		pwdchange = false;

		if (type == 1)
		{
			memset(pcUrl, 0, IOTA_BUF_SIZE_64);
			getUrl_v3(ip, pcUrl, port);
			dzlog_info("mqtt_login new ip_list.json pcUrl[%s]", pcUrl);
		}
		else
		{
			memset(pcUrl, 0, IOTA_BUF_SIZE_64);
			if (iden != 0)
			{
				if (g_uiPort != 18830)
				{
					getUrl_v1(ip, pcUrl, 1);
				}
				else if (g_mqtt_issucc == -5)
				{
					getUrl_v1(ip, pcUrl, 1);
				}
				else
				{
					return -3;
				}
			}
			else
			{
				getUrl_v1(ip, pcUrl, 0);
			}
		}
		// sprintf(pcUrl, "ssl://%s:%d", "10.235.2.103", 8883);
		printf("start to connect mqtt server:%s,id:%s\n", pcUrl, g_pcClientId);
		MQTTAsync_createOptions create_opts = MQTTAsync_createOptions_initializer;
		if(g_mqtt_ob.clientInfo.scram == 1)
		{
			MQTTAsync_createOptions create_opts1 = MQTTAsync_createOptions_initializer5;
			create_opts = create_opts1;
		}

		// int iRet = MQTTAsync_create(&g_client, pcUrl, g_pcClientId, MQTTCLIENT_PERSISTENCE_NONE, NULL);
		int iRet = MQTTAsync_createWithOptions(&g_client, pcUrl, g_pcClientId, MQTTCLIENT_PERSISTENCE_NONE, NULL, &create_opts);
		if (iRet != MQTTASYNC_SUCCESS)
		{
			dzlog_error("Failed to create client, return code %d\n", iRet);
			bConnect = 0;
			IOTA_destroy();
			return iRet;
		}
		// 设置登录后，接收消息回调
		iRet = MQTTAsync_setCallbacks(g_client, NULL, connlost, msgarrvd, delivered);
		if (iRet != MQTTASYNC_SUCCESS)
		{
			dzlog_error("Failed to set callbacks, return code %d\n", iRet);
			bConnect = 0;
			IOTA_destroy();
			return iRet;
		}
		MQTTAsync_SSLOptions ssl_opts = MQTTAsync_SSLOptions_initializer;
		if (g_mqtt_ob.clientInfo.SSL == 1)					//采用标准SSL
		{
			// 单向认证
			ssl_opts.sslVersion = MQTT_SSL_VERSION_TLS_1_2; // 指定 TLS 版本
			ssl_opts.trustStore = CLIENT_CA_CERT_PATH_V1;	// CA 证书路径
			ssl_opts.keyStore = CLIENT_SIGN_CERT_PATH_V1;	// 客户端证书路径
			ssl_opts.privateKey = CLIENT_SIGN_KEY_PATH_V1;	// 客户端私钥路径
			ssl_opts.dkeyStore = NULL;
			ssl_opts.dprivateKey = NULL;
			ssl_opts.enableServerCertAuth = 0; // 启用服务端证书验证
		}
		else
		{
			// 不进行认证
			ssl_opts.trustStore = strlen(g_pcCertPath) == 0 ? "./mqttClientTrustCert.pem" : g_pcCertPath;
			ssl_opts.sslVersion = MQTT_SSL_VERSION_DEFAULT;
		}

		MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
		if(g_mqtt_ob.clientInfo.scram == 1)
		{
		}
		else
		{
			char tmp_pcUserName[64], tmp_pcPassword[64];
			conn_opts.keepAliveInterval = g_mqtt_ob.clientInfo.keepAlivetime;
			// 连接成功回调处理
			conn_opts.onSuccess = onConnectSuccess;
			// 连接失败回调处理
			conn_opts.onFailure = onConnectFailure;

			if (type == 1)
			{
				pwdchange = true;
				conn_opts.username = user;
				conn_opts.password = pwd;
			}
			else
			{
				int tmp = getconninfo(ip, 0, tmp_pcUserName);
				if (tmp != -1)
				{
					getconninfo(ip, 1, tmp_pcPassword);
					pwdchange = true;
					conn_opts.username = tmp_pcUserName;
					conn_opts.password = tmp_pcPassword;
				}
				else
				{
					conn_opts.username = g_pcUserName;
					conn_opts.password = g_pcPassword;
				}

			}
		}

		conn_opts.connectTimeout = 8;
		conn_opts.cleansession = 1;
		conn_opts.context = NULL;
		conn_opts.automaticReconnect = 1; // 设置非零，断开自动重连
		conn_opts.minRetryInterval = 4;	  // 单位秒，重连间隔时间，每次重新连接失败时，重试间隔都会加倍，直到最大间隔
		conn_opts.maxRetryInterval = 32;  // 单位秒，最大重连尝试间隔
		// ssl选项
		if (g_mqtt_ob.clientInfo.SSL == 0)
			conn_opts.ssl = NULL;
		else
			conn_opts.ssl = &ssl_opts;

		g_mqtt_issucc = -1; // 初始为-1
		iRet = MQTTAsync_setConnected(g_client, NULL, onConnectCallCBack);
		if (iRet != MQTTASYNC_SUCCESS)
		{
			printf("Failed to set callback, return code %d\n", iRet);
			dzlog_info("Failed to set callback, return code %d\n", iRet);
			bConnect = 0;
			IOTA_destroy();
			return iRet;
		}
		dzlog_info("start to connect mqtt server:%s,g_uiSSL:%d,id:%s", pcUrl, g_mqtt_ob.clientInfo.SSL, g_pcClientId);
		iRet = MQTTAsync_connect(g_client, &conn_opts);
		if (iRet != MQTTASYNC_SUCCESS)
		{
			dzlog_error("Failed to connect, return code %d\n", iRet);
			bConnect = 0;
			//IOTA_logout();
			IOTA_destroy();
			return iRet;
		}
		bConnect = 1;

		// 这里测试时发现，可能不对的ip，MQTTAsync_connect接口也能成功，所以增加g_mqtt_issucc标识。
		// 又在仿真环境测试时，发现g_mqtt_issucc置为1(在mqtt连接成功的回调里置1)可能需要5秒以上，所以要sleep;
		int ii = 0;
		for (ii = 0; ii < 8; ii++)
		{
			if (g_mqtt_issucc == 1)
			{
				// 成功连接后，更新缓存信息，比如ip  priorityconf
				dzlog_info("mqtt Connection ssucc:%d pcUrl[%s]", g_mqtt_issucc, pcUrl);
				http_init_url(ip, SSL);
				
				if (pwdchange)
				{
					strncpy(g_pcUserName, conn_opts.username, IOTA_BUF_SIZE_64 - 1);
					strncpy(g_pcPassword, conn_opts.password, IOTA_BUF_SIZE_64 - 1);
					dzlog_info("IOTA_login pwdchange g_pcUserName[%s],g_pcPassword[%s]", g_pcUserName, g_pcPassword);
				}

				return 0;
			}
			else if (g_mqtt_issucc == -2)
			{
				// g_mqtt_issucc == -2是连接失败回调里设置为-2
				// dzlog_info("mqtt Connection error:%d pcUrl[%s]", g_mqtt_issucc, pcUrl);
				IOTA_destroy();
				break;
			}
			else if (g_mqtt_issucc == -5)
			{
				// g_mqtt_issucc == -2是连接失败回调里设置为-2
				// dzlog_info("mqtt Connection error:%d pcUrl[%s]", g_mqtt_issucc, pcUrl);
				IOTA_destroy();
				break;
			}
			printf("mqtt Connection failed:%d pcUrl[%s]\n", g_mqtt_issucc, pcUrl);
			sleep(1);
		}

		if (g_mqtt_issucc != 1)
		{
			dzlog_info("mqtt Connection error:%d pcUrl[%s]", g_mqtt_issucc, pcUrl);
			printf("mqtt Connection error:%d pcUrl[%s]\n", g_mqtt_issucc, pcUrl);
			//IOTA_logout();
			IOTA_destroy();
			if (g_mqtt_issucc == -5 || g_mqtt_issucc == -2)
			{
				continue;
			}
			else
			{
				return -1;
			}
		}
	}
	return -1;
}

int IOTA_login()
{
	int iRet = IOTA_SUCCESS;
	iRet = IOTA_login_fromiplist();
	return iRet;
}

int update_connection_profile(const char *filename, char *cli_id, char *cli_ip, char *cli_port, char *cli_user, char *cli_password, int ssl, char* edgeNodeId)
{
	// 组装json

	cJSON *root = cJSON_CreateObject();
	if (root)
	{
		if (edgeNodeId != NULL)
		{
			cJSON_AddStringToObject(root, "edgeNodeId", edgeNodeId);
		}
		
		cJSON_AddStringToObject(root, "client_id", cli_id);
		cJSON_AddStringToObject(root, "host", cli_ip);
		cJSON_AddNumberToObject(root, "port", atoi(cli_port));
		cJSON_AddStringToObject(root, "username", cli_user);
		cJSON_AddStringToObject(root, "password", cli_password);
		cJSON_AddNumberToObject(root, "tls", ssl);
	}
	else
	{
		return -1;
	}

	// 写回文件
	FILE *fp = fopen(filename, "w");
	if (!fp)
	{
		cJSON_Delete(root);
		return -3;
	}

	char *json_str = cJSON_Print(root);
	fputs(json_str, fp);

	fclose(fp);
	cJSON_Delete(root);
	free(json_str);

	return 0;
}


int IOTA_login_fromiplist()
{
	int iRet = IOTA_SUCCESS;

	int bCreate = 0, bSet = 0, bConnect = 0;
	char pcJson[1024 * 15] = { 0 };

	while (true)
	{
		int ii = 0;
		char dev_ip[32] = "";
		char port[16] = "";
		char username[32] = "";
		char password[32] = "";
		//增加网络测试，网络通再执行
		int result = check_connectivity(dev_ip, atoi(port), 5);
		dzlog_info("222111 dev_ip[%s], port[%s],result[%d]", dev_ip, port, result);
		if (result != 0)
		{
			continue;
		}
		smc_tls = ii;
		dzlog_info("smc_tls[%d]", smc_tls);
		int res = mqtt_login(dev_ip, 1, port, username, password, ii);
		if (res == 0)
		{
			// 成功
			bConnect = 1;

			//将连接配置更新到外部配置文件Update Profile
			update_connection_profile("./conf/smc_connection.json", g_pcClientId, dev_ip, port, username, password, ii, "000000");
			memset(g_ServerPort, 0, 64);
			memset(g_ServerIp, 0, 64);
			sprintf(g_ServerPort, "%s", port);
			sprintf(g_ServerIp, "%s", dev_ip);
			break;
		}

		if (bConnect && g_mqtt_issucc)
		{
			break;
		}
		else
		{
			sleep(30);
			continue;
		}
	}
	return iRet;
}

int IOTA_logout()
{
	int iRet = IOTA_SUCCESS;

	MQTTAsync_disconnectOptions disconn_opts = MQTTAsync_disconnectOptions_initializer;
	if(g_mqtt_ob.clientInfo.scram == 1)
	{
		MQTTAsync_disconnectOptions disconn_opts1 = MQTTAsync_disconnectOptions_initializer5;
		disconn_opts = disconn_opts1;
		disconn_opts.onSuccess5 = onDisconnectSuccess;
		disconn_opts.onFailure5 = onDisconnectFailure;
	}
	else
	{
		disconn_opts.onSuccess = onDisconnectSuccess;
		disconn_opts.onFailure = onDisconnectFailure;
	}

	disconn_opts.context = NULL;
	iRet = MQTTAsync_disconnect(g_client, &disconn_opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		dzlog_error("Failed to disconnect, return code %d\n", iRet);
	}

	return iRet;
}

int IOTA_hubDeviceAdd(HW_UINT uiMid, ST_IOTA_DEVICE_INFO *pstInfo)
{
	int iRet = IOTA_SUCCESS;

	char *pcJson = NULL;
	if (StructToJson_DeviceAdd(uiMid, pstInfo, &pcJson))
	{
		publishMessage(g_pcTopicTopoAdd, pcJson);
	}
	else
	{
		iRet = IOTA_FAILURE;
	}
	free_s(pcJson);
	return iRet;
}

int IOTA_hubDeviceRemove(HW_UINT uiMid, char *pcDeviceId)
{
	int iRet = IOTA_SUCCESS;

	char *pcJson = NULL;
	if (StructToJson_DeviceRemove(uiMid, pcDeviceId, &pcJson))
	{
		publishMessage(g_pcTopicTopoDelete, pcJson);
	}
	else
	{
		iRet = IOTA_FAILURE;
	}
	free_s(pcJson);
	return iRet;
}

int IOTA_deviceStatusUpdate(HW_UINT uiMid, char *pcDeviceId, char *pcStatus)
{
	int iRet = IOTA_SUCCESS;

	char *pcJson = NULL;
	ST_IOTA_DEVICE_STATUS status;
	status.pcDeviceId = pcDeviceId;
	status.pcStatus = pcStatus;
	if (StructToJson_DeviceStatusUpdate(uiMid, &status, &pcJson))
	{
		publishMessage(g_pcTopicTopoUpdate, pcJson);
	}
	else
	{
		iRet = IOTA_FAILURE;
	}
	free_s(pcJson);
	return iRet;
}

int IOTA_hubDeviceQuery(HW_UINT uiMid, char *pcMarker, HW_UINT uiLimit, ST_IOTA_DEVICE_INFO *pstDeviceInfo)
{
	int iRet = IOTA_SUCCESS;

	char *pcJson = NULL;
	if (StructToJson_DeviceQuery(uiMid, pstDeviceInfo->pcNodeId, &pcJson))
	{
		publishMessage(g_pcTopicTopoQuery, pcJson);
	}
	else
	{
		iRet = IOTA_FAILURE;
	}
	free_s(pcJson);
	return iRet;
}

int IOTA_serviceDataReport(char *pcDeviceId, char *pcServiceId, char *pcServiceProperties)
{
	int iRet = IOTA_SUCCESS;

	char *pcJson = NULL;
	if (StructToJson_ServiceDataReport(pcDeviceId, pcServiceId, pcServiceProperties, &pcJson))
	{
		publishMessage(g_pcTopicDatas, pcJson);
	}
	else
	{
		iRet = IOTA_FAILURE;
	}
	free_s(pcJson);
	return iRet;
}

int IOTA_serviceCommandResponse(HW_UINT uiMid, HW_UINT uiResultCode, char *pcCommandResponse)
{
	int iRet = IOTA_SUCCESS;

	char *pcJson = NULL;
	if (StructToJson_ServiceCommandResponse(uiMid, uiResultCode, pcCommandResponse, &pcJson))
	{
		publishMessage(g_pcTopicCommandResponse, pcJson);
	}
	else
	{
		iRet = IOTA_FAILURE;
	}
	free_s(pcJson);
	return iRet;
}

int IOTA_serviceCustomTopicReport(ST_IOTA_BatchReportProperties *pstDevInfo, char *pcTopicVersion, char *pcCustomTopic)
{
	int iRet = IOTA_SUCCESS;

	if (pcTopicVersion == NULL || pcCustomTopic == NULL || pstDevInfo == NULL)
	{
		iRet = IOTA_PARAMETER_EMPTY;
		return iRet;
	}

	char *pcJson = NULL;
	char buf[256] = {0};
	snprintf(buf, 255, "/%s/%s/%s", pcTopicVersion, g_pcDeviceId, pcCustomTopic);
	if (StructToJson_ServiceCustomTopicReport(pstDevInfo, &pcJson))
	{
		publishMessage(buf, pcJson);
	}
	else
	{
		iRet = IOTA_FAILURE;
	}
	free_s(pcJson);
	return iRet;
}

int IOTA_subscribeCustomTopic(char *pcTopicVersion, char *pcCustomTopic)
{
	int iRet = IOTA_SUCCESS;

	if (pcTopicVersion == NULL || pcCustomTopic == NULL)
	{
		iRet = IOTA_PARAMETER_EMPTY;
		return iRet;
	}

	char buf[256] = {0};
	snprintf(buf, 255, "/%s/%s/%s", pcTopicVersion, g_pcDeviceId, pcCustomTopic);

	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	if(g_mqtt_ob.clientInfo.scram == 1)
	{
		opts.onSuccess5 = onSubscribeSuccess;
		opts.onFailure5 = onSubscribeFailure;
	}
	else
	{
		opts.onSuccess = onSubscribeSuccess;
		opts.onFailure = onSubscribeFailure;
	}

	opts.context = NULL;
	iRet = MQTTAsync_subscribe(g_client, buf, g_uiQos, &opts);
	if (iRet != MQTTASYNC_SUCCESS)
	{
		dzlog_error("Failed to subscribe, return code %d\n", iRet);
	}
	else
	{
		dzlog_info("MQTTAsync_subscribe, topic: %s\n", buf);
	}
	return iRet;
}

int IOTA_setCertPath(char *pcCertPath)
{
	int iRet = IOTA_SUCCESS;
	strncpy(g_pcCertPath, pcCertPath, IOTA_BUF_SIZE_512 - 1);
	return iRet;
}

void getUrl(char *pcUrl)
{
	if (pcUrl == NULL)
	{
		return;
	}

	if (g_uiSSL == 0)
	{
		sprintf(pcUrl, "tcp://%s:%d", g_pcServerAddr, g_uiPort);
	}
	else
	{
		sprintf(pcUrl, "ssl://%s:%d", g_pcServerAddr, g_uiPort);
	}

	return;
}

/*
getUrl_v1(input char* ip, output char* pcUrl)
输入ip,返回mqtt的url
*/
void getUrl_v1(char *ip, char *pcUrl, int tailor)
{
	if (pcUrl == NULL || ip == NULL)
	{
		return;
	}

	if (g_uiSSL == 0)
	{
		if (g_uiPort != 18830 && tailor == 1)
		{
			sprintf(pcUrl, "tcp://%s:%d", ip, 18830);
	}
		else
		{
			sprintf(pcUrl, "tcp://%s:%d", ip, g_uiPort);
		}
	}
	else
	{
		if (g_uiPort != 18830 && tailor == 1)
		{
			sprintf(pcUrl, "ssl://%s:%d", ip, 18830);
		}
		else
		{
			sprintf(pcUrl, "ssl://%s:%d", ip, g_uiPort);
		}
	}
	return;
}

void getUrlV2(char *ip, char *pcUrl, uint16_t iPort)
{
	if (pcUrl == NULL || ip == NULL)
	{
		return;
	}

	if (g_uiSSL == 0)
	{
		sprintf(pcUrl, "tcp://%s:%d", ip, iPort);
	}
	else
	{
		sprintf(pcUrl, "ssl://%s:%d", ip, iPort);
	}
	return;
}

/*
getUrl_v3(char* ip, char* pcUrl, int port)
输入ip,返回mqtt的url
*/
void getUrl_v3(char *ip, char *pcUrl, char *port)
{
	if (pcUrl == NULL || ip == NULL)
	{
		return;
	}

	if (g_uiSSL == 0)
	{
		sprintf(pcUrl, "tcp://%s:%s", ip, port);
	}
	else if (g_uiSSL == 1)
	{
		sprintf(pcUrl, "ssl://%s:%s", ip, port);
	}
	else
	{
		sprintf(pcUrl, "ssl://%s:%s", ip, port);
	}
	return;
}
